package com.bbzn.device.client.config;


import com.bbzn.device.client.enums.TopicName;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;

import javax.annotation.Resource;
import java.util.Arrays;

/**
 * @Author wulongbo
 * @Date 2020/12/29 11:38
 * @Version 1.0
 *
 * EMQX配置工具类
 */

@Configuration
@IntegrationComponentScan //消息扫描件
@Slf4j
public class EmqxMqttConfig {

    @Resource
    private EmqxMqttProperties emqxMqttProperties;

    @Bean
    public DefaultMqttPahoClientFactory getMqttConnectOptions() {

        DefaultMqttPahoClientFactory defaultMqttPahoClientFactory = new DefaultMqttPahoClientFactory();
        defaultMqttPahoClientFactory.setUserName(emqxMqttProperties.getUsername());
        defaultMqttPahoClientFactory.setPassword(emqxMqttProperties.getPassword());
        defaultMqttPahoClientFactory.setServerURIs(new String[]{emqxMqttProperties.getHostUrl()});
        // 心跳
        defaultMqttPahoClientFactory.setKeepAliveInterval(emqxMqttProperties.getKeepAlive());
        defaultMqttPahoClientFactory.setConnectionTimeout(emqxMqttProperties.getTimeout());
        // 保留/清空曾经连接的客户端信息
        defaultMqttPahoClientFactory.setCleanSession(false);
        // qos
        String playload = "设备已断开连接";
        // 遗嘱消息
        defaultMqttPahoClientFactory.setWill(new DefaultMqttPahoClientFactory.Will("last_topic", playload.getBytes(), emqxMqttProperties.getQos(), false));
        return defaultMqttPahoClientFactory;
    }

    @Bean
    public MessageChannel mqttSubscribeChannel() {
        return new DirectChannel();
    }

    @Bean
    public MessageChannel nbMqttSubscribeChannel() {
        return new DirectChannel();
    }

    @Bean
    public MessageChannel bridgeMqttSubscribeChannel() {
        return new DirectChannel();
    }

    /**
     * 配置client,监听的topic
     */
    @Bean
    public MessageProducer inbound() {
        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter(emqxMqttProperties.getClientId() + "_inbound", getMqttConnectOptions(),
                        emqxMqttProperties.getDefaultTopic().split(","));
        adapter.setCompletionTimeout(Math.toIntExact(Long.valueOf(emqxMqttProperties.getTimeout())));
        adapter.setConverter(new DefaultPahoMessageConverter());
        //默认添加TopicName中所有tipic
        // 当一个client处理的时候使用遍历同意订阅
//        Arrays.stream(TopicName.values()).forEach(topicName -> adapter.addTopic(topicName.getValue(), 2));

        // 当使用多个client处理的时候单独处理
        adapter.addTopic(TopicName.ROLL_CALL_DEFAULT.getValue(),2);
//        Arrays.stream(emqxMqttProperties.getDefaultTopic().split(",")).forEach(topicName -> adapter.addTopic(topicName, 2));
//        adapter.addTopic();
        adapter.setQos(2);
        adapter.setOutputChannel(mqttSubscribeChannel());
        return adapter;
    }



    /**
     * 配置nb的client,监听的topic  nb/imei/client
     */
    @Bean
    public MessageProducer nbInbound() {
        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter(emqxMqttProperties.getClientId() + "_nbInbound", getMqttConnectOptions(),
                        emqxMqttProperties.getNbTopic().split(","));
        adapter.setCompletionTimeout(Math.toIntExact(Long.valueOf(emqxMqttProperties.getTimeout())));
        adapter.setConverter(new DefaultPahoMessageConverter());
        //默认添加TopicName中所有tipic
        // 当一个client处理的时候使用遍历同意订阅
//        Arrays.stream(TopicName.values()).forEach(topicName -> adapter.addTopic(topicName.getValue(), 2));
        // 当使用多个client处理的时候单独处理
        adapter.addTopic(TopicName.NB_TOPIC_DEFAULT.getValue(),2);

//        Arrays.stream(emqxMqttProperties.getDefaultTopic().split(",")).forEach(topicName -> adapter.addTopic(topicName, 2));
//        adapter.addTopic();
        adapter.setQos(2);
        adapter.setOutputChannel(nbMqttSubscribeChannel());
        return adapter;
    }


    /**
     * 配置桥接的client,监听的topic  bridge/imei/client
     */
    @Bean
    public MessageProducer bridgeInbound() {
        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter(emqxMqttProperties.getClientId() + "_bridgeInbound", getMqttConnectOptions(),
                        emqxMqttProperties.getBridgeTopic().split(","));
        adapter.setCompletionTimeout(Math.toIntExact(Long.valueOf(emqxMqttProperties.getTimeout())));
        adapter.setConverter(new DefaultPahoMessageConverter());
        //默认添加TopicName中所有tipic
        // 当一个client处理的时候使用遍历同意订阅
//        Arrays.stream(TopicName.values()).forEach(topicName -> adapter.addTopic(topicName.getValue(), 2));
        // 当使用多个client处理的时候单独处理
        adapter.addTopic(TopicName.BRIDGE_TOPIC_DEFAULT.getValue(),2);

//        Arrays.stream(emqxMqttProperties.getDefaultTopic().split(",")).forEach(topicName -> adapter.addTopic(topicName, 2));
//        adapter.addTopic();
        adapter.setQos(2);
        adapter.setOutputChannel(bridgeMqttSubscribeChannel());
        return adapter;
    }


    /**
     * 事件触发
     */
    @Autowired
    private ApplicationEventPublisher eventPublisher;


    @Bean
    @ServiceActivator(inputChannel = Constants.MQTT_PUBLISH_CHANNEL)
    public MessageHandler mqttOutbound() {

        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(emqxMqttProperties.getClientId(), getMqttConnectOptions());
        messageHandler.setAsync(true);
        messageHandler.setDefaultTopic(emqxMqttProperties.getDefaultTopic());
        return messageHandler;

    }


    @Bean
    @ServiceActivator(inputChannel = Constants.MQTT_NB_PUBLISH_CHANNEL)
    public MessageHandler nbMqttOutbound() {

        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(emqxMqttProperties.getClientId(), getMqttConnectOptions());
        messageHandler.setAsync(true);
        messageHandler.setDefaultTopic(emqxMqttProperties.getNbTopic());
        return messageHandler;

    }

    @Bean
    @ServiceActivator(inputChannel = Constants.MQTT_BRIDGE_PUBLISH_CHANNEL)
    public MessageHandler bridgeMqttOutbound() {
        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(emqxMqttProperties.getClientId(), getMqttConnectOptions());
        messageHandler.setAsync(true);
        messageHandler.setDefaultTopic(emqxMqttProperties.getBridgeTopic());
        return messageHandler;

    }

    @Bean
    public MessageChannel mqttPublishChannel() {

        return new DirectChannel();
    }

    @Bean
    public MessageChannel nbMqttPublishChannel() {

        return new DirectChannel();
    }

    @Bean
    public MessageChannel bridgeMqttPublishChannel() {
        return new DirectChannel();
    }



}
